package com.hrt.flink.icebergoperate.unauto;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *  Flink SQL API 实时读取Kafka 写入到Iceberg
 */
public class FlinkSQLIceberg3 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);

        Configuration configuration = tblEnv.getConfig().getConfiguration();
        // 支持SQL语法中的 OPTIONS 选项
        configuration.setBoolean("table.dynamic-table-options.enabled", true);

        env.enableCheckpointing(1000);

        //1.创建Catalog
        tblEnv.executeSql("create catalog hadoop_iceberg with ('type'='iceberg','catalog-type'='hadoop','warehouse'='hdfs://mycluster/flink_iceberg')");

        //2.创建Iceberg表
//        tblEnv.executeSql("create table hadoop_iceberg.iceberg_db.flink_iceberg_tbl3 (id int,name string,age int,loc string) partitioned by (loc)");

        //3.读取Kafka 中的数据
        tblEnv.executeSql("create table kafka_input_table(" +
                "id int," +
                "name varchar," +
                "age int," +
                "loc varchar" +
                ") with (" +
                "'connector'='kafka'," +
                "'topic' = 'flink-iceberg-topic'," +
                "'properties.bootstrap.servers'='node1:9092,node2:9092,node3:9092'," +
                "'scan.startup.mode'='latest-offset'," +
                "'format'='csv'," +
                "'properties.group.id'='my-group-id'" +
                ")");

        //4.向Iceberg表中写数据
        tblEnv.executeSql("insert into hadoop_iceberg.iceberg_db.flink_iceberg_tbl3 select id,name,age,loc from kafka_input_table");

        //5.查询Iceberg表数据并打印
        TableResult tableResult = tblEnv.executeSql("SELECT * FROM hadoop_iceberg.iceberg_db.flink_iceberg_tbl3 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/");
        tableResult.print();

    }
}
